package com.niodata.dp.monitor.components.task.service;

import com.google.common.base.Strings;
import com.niodata.dp.monitor.components.task.dao.TaskAlertMsgMapper;
import com.niodata.dp.monitor.components.task.dao.TaskExecutionMapper;
import com.niodata.dp.monitor.core.task.TaskExecutionManager;
import com.niodata.dp.monitor.core.task.az.FlowExecutionDetail;
import com.niodata.dp.monitor.sharelib.entity.az.AzkabanSummary;
import com.niodata.dp.monitor.sharelib.entity.az.FlowExecutor;
import com.niodata.dp.monitor.sharelib.entity.task.TaskAlertMsg;
import com.niodata.dp.monitor.sharelib.entity.task.TaskExecResult;
import com.niodata.dp.monitor.sharelib.entity.task.TaskExecution;
import com.niodata.dp.monitor.sharelib.entity.task.TaskMonitorResult;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;


/**
 * Created by ndp on 17-7-10.
 */
@Service
public class TaskAlertService {


  @Autowired
  TaskAlertMsgMapper taskAlertMsgMapper;

  @Autowired
  TaskExecutionMapper taskExecutionMapper;

  @Autowired
  @Qualifier("taskManagerImpl")
  TaskExecutionManager taskManager;


  SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

  // 1970-01-02 00:00:00
  // it is a database bug
  // mysql> ALTER TABLE TASK_EXECUTION MODIFY
  // COLUMN END_TIME timestamp not null default '1970-01-01 00:00:00';
  // ERROR 1067 (42000): Invalid default value for 'END_TIME'
  private long defaultMinTime = 24 * 3600;

  /**
   * taskAlertMsgsFilter.
   */
  public TaskMonitorResult taskAlertMsgsFilter(long userId, String groupIds, int pageIndex,
        int pageSize, String taskId, String status) {
    int offset = (pageIndex - 1) * pageSize < 0 ? 0 : (pageIndex - 1) * pageSize;
    List<Long> ids = new ArrayList<>();
    if (!Strings.isNullOrEmpty(groupIds)) {
      ids = Arrays.stream(groupIds.trim().split(",")).map(str -> {
        return Long.valueOf(str);
      }).collect(Collectors.toList());
    }
    List<TaskAlertMsg> result = taskAlertMsgMapper
          .getTaskAlertMsgsByOwner(userId, ids, offset, pageSize + 1, Strings.emptyToNull(taskId),
                Strings.emptyToNull(status));

    TaskMonitorResult monitorResult = new TaskMonitorResult();
    monitorResult.setHasMore(result.size() > pageSize);
    monitorResult.setPageIndex(pageIndex);
    monitorResult.setStart(pageIndex <= 1 && result.size() == 0 ? 0 : offset + 1);
    monitorResult.setStop(pageSize < result.size() ? offset + pageSize : offset + result.size());
    if (!result.isEmpty()) {
      result.remove(result.size() - 1);
    }
    monitorResult.setData(result);
    monitorResult.setSuccess(true);
    monitorResult.setTimestamp(System.currentTimeMillis());
    return monitorResult;
  }

  /**
   * status,sperated by comma,like SUCCESSED,KILLED etc. getTaskExecResult.
   */
  public TaskExecResult getTaskExecResult(String taskId, long userId, List<Long> ids, int pageIndex,
        int pageSize, String status,
        String startTime, String endTime) {
    int offset = (pageIndex - 1) * pageSize < 0 ? 0 : (pageIndex - 1) * pageSize;
    List<TaskExecution> result = null;
    if (StringUtils.isBlank(startTime)) {
      startTime = null;
    }
    if (StringUtils.isBlank(endTime)) {
      endTime = null;
    }
    if (status != null && !status.isEmpty()) {
      result = taskExecutionMapper
            .getTaskExecutionByOwner(taskId, userId, ids, offset, pageSize + 1, status, startTime,
                  endTime);
    } else {
      result = taskExecutionMapper
            .getAllTaskExecutionByOwner(taskId, userId, ids, offset, pageSize + 1, startTime,
                  endTime);
    }

    TaskExecResult taskExecResult = new TaskExecResult();
    taskExecResult.setHasMore(result.size() > pageSize);
    taskExecResult.setPageIndex(pageIndex);
    taskExecResult.setStart(pageIndex <= 1 && result.size() == 0 ? 0 : offset + 1);
    taskExecResult.setStop(pageSize < result.size() ? offset + pageSize : offset + result.size());
    if (!result.isEmpty() && result.size() > pageSize) {
      result.remove(result.size() - 1);
    }
    taskExecResult.setData(result);
    taskExecResult.setSuccess(true);
    taskExecResult.setTimestamp(System.currentTimeMillis());
    return taskExecResult;
  }

  /**
   * get recently finished executions .
   *
   * @param userId userId
   * @param groupIds groupIds
   * @return taskexec result
   */
  public TaskExecResult getRecentlyFinishedExec(long userId, List<Long> groupIds) {

    Date minDate = DateUtils.addMinutes(new Date(), -15);
    List<TaskExecution> executions = taskExecutionMapper.getRecentlyFinishedExecsByUser(userId,
          groupIds, minDate);
    TaskExecResult taskExecResult = new TaskExecResult();
    taskExecResult.setHasMore(false);
    taskExecResult.setPageIndex(1);
    taskExecResult.setStart(0);
    taskExecResult.setStop(executions.size());
    taskExecResult.setData(executions);
    taskExecResult.setSuccess(true);
    taskExecResult.setTimestamp(System.currentTimeMillis());
    return taskExecResult;
  }

  public Map<String, Long> getAzKaBanStatus(Date start, Date stop) {
    List<TaskExecution> taskExecutions = taskExecutionMapper
          .getTaskExecutionByTime(sdf.format(start), sdf.format(stop));

    return taskExecutions.stream().map(TaskExecution::getStatus)
          .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));
  }

  public void updateLastExecStatus() {
    List<TaskExecution> taskExecutions = taskExecutionMapper.getRunningTaskExecution();
    taskExecutions.forEach(t -> {
      FlowExecutionDetail detail = taskManager.getTaskExecutionDetail(t.getExecId());
      taskExecutionMapper.update(detail.getExecid(), detail.getStatus(),
            detail.getStartTime(), detail.getUpdateTime());
    });
  }

  /**
   * getAzkabanSummary.
   */
  public AzkabanSummary getAzkabanSummary(Date startTime, Date stopTime) {
    Map<String, Long> counter = getAzKaBanStatus(startTime, stopTime);
    Long scheduled = taskManager.getAllTasks().stream().filter(taskInfo -> {
      return taskInfo.isScheduled();
    }).count();
    counter.put("Scheduled", scheduled);
    AzkabanSummary azkabanSummary = new AzkabanSummary();
    try {
      List<FlowExecutor> flowExecutors = taskManager.getFlowExecutors();
      azkabanSummary.setFlowExecutorList(flowExecutors);

    } catch (IOException ioe) {
      ioe.printStackTrace();
    }
    azkabanSummary.setSummary(counter);
    return azkabanSummary;
  }
}
